package net.bwie.realtime.jtp.utils;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class jianbiao {
    public static void main(String[] args) throws Exception {
        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 创建创表环境
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        tEnv.executeSql("CREATE TABLE douyin_live_business (\n" +
                "  `orderId` STRING,\n" +
                "  `eventTime` STRING,\n" +
                "  `roomId` STRING,\n" +
                "  `userId` STRING,\n" +
                "  `productSku` STRING,\n" +
                "  `orderAmount` FLOAT,\n" +
                "  `payAmount` STRING,\n" +
                "  `payStatus` STRING,\n" +
                "  `anchorId` STRING,\n" +
                "  `roomType` STRING,\n" +
                "  `startTime` STRING,\n" +
                "  `endTime` STRING,\n" +
                "  `interactId` STRING,\n" +
                "  `interactEventType` STRING,\n" +
                "  `interactType` STRING,\n" +
                "  `content` STRING,\n" +
                "  `visitId` STRING,\n" +
                "  `enterTime` STRING,\n" +
                "  `leaveTime` STRING,\n" +
                "  `source` STRING,\n" +
                "  `duration` INT,\n" +
                "  `productName` STRING,\n" +
                "  `price` DOUBLE,\n" +
                "  `category` STRING,\n" +
                "  `status` STRING \n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'douyin_live_business',\n" +
                "  'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                "  'properties.group.id' = 'testGroup',\n" +
                "  'scan.startup.mode' = 'earliest-offset',\n" +
                "  'format' = 'json'\n" +
                ")");
        tEnv.executeSql("select * from douyin_live_business limit 10").print();
        /*

         */

        tEnv.sqlQuery("SELECT window_start, window_end, SUM(price)\n" +
                "  FROM TABLE(\n" +
                "    HOP(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES, INTERVAL '10' MINUTES))\n" +
                "  GROUP BY window_start, window_end;").execute().print();


        tEnv.sqlQuery("SELECT window_start, window_end, SUM(price)\n" +
                "  FROM TABLE(\n" +
                "    HOP(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES, INTERVAL '10' MINUTES))\n" +
                "  GROUP BY window_start, window_end;").execute().print();








        // 启动
        env.execute("jianbiao");
    }
}
